package com.demo.kafkalearn.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class Producer {

    private static final String brokerList = "192.168.13.177:9092";
    private static final String topic = "topic-demo";

    public static void main(String[] args) {
        Properties properties = new Properties();
        //设置key序列化器
        //properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //设置重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG, 10);
        //properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //设置集群地址
        //properties.put("bootstrap.servers", brokerList);
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        //创建生产者客户端实例
        KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String,String>(properties);
        //构建所需要发送的消息
        ProducerRecord<String,String> record = new ProducerRecord<>(topic,"kafka-demo","hello,kafka!");
        try{
            //发送消息
            kafkaProducer.send(record);
        }catch (Exception e){
            e.printStackTrace();
        }
        //关闭生产者客户端示例
        kafkaProducer.close();
    }
}
